package com.mrx.mqtt.mqttintegration.config;

import com.mrx.mqtt.mqttintegration.mqtt.service.MqttService;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;

import javax.annotation.PostConstruct;

/**
 * @author XH.M
 * @create 2019-04-28 10:53
 */
@Configuration
@IntegrationComponentScan
public class MqttConfig {

    @Autowired
    private MqttService mqttService;

    @Value("${mqtt.username}")
    private String username;
    @Value("${mqtt.password}")
    private String password;
    @Value("${mqtt.url}")
    private String url;
    @Value("${mqtt.clientId}")
    private String clientId;
    @Value("${mqtt.defaultTopic}")
    private String defaultTopic;
    @Value("${mqtt.keepAliveInterval}")
    private Integer keepAliveInterval;
    @Value("${mqtt.automaticReconnect}")
    private Boolean automaticReconnect;
    @Value("${mqtt.cleanSession}")
    private Boolean cleanSession;
    @Value("${mqtt.completionTimeout}")
    private Integer completionTimeout;

    @PostConstruct
    public void outPut(){
        System.out.println("username: "+username);
        System.out.println("password: "+password);
        System.out.println("url: "+url);
        System.out.println("clientId: "+clientId);
        System.out.println("defaultTopic: "+defaultTopic);
        System.out.println("keepAliveInterval: "+keepAliveInterval);
        System.out.println("automaticReconnect: "+automaticReconnect);
        System.out.println("cleanSession: "+cleanSession);
        System.out.println("completionTimeout: "+completionTimeout);
    }

    @Bean
    public MqttConnectOptions mqttConnectOptions(){
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setServerURIs(new String[]{url});
        options.setKeepAliveInterval(keepAliveInterval);
        options.setAutomaticReconnect(automaticReconnect);
        options.setCleanSession(cleanSession);
        options.setConnectionTimeout(completionTimeout);
        return options;
    }

    @Bean
    public MqttPahoClientFactory mqttPahoClientFactory(MqttConnectOptions mqttConnectOptions){
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(mqttConnectOptions);
        return factory;
    }

    // 接收通道
    @Bean
    public MessageChannel mqttInputChannel(){
        return new DirectChannel();
    }

    // 配置client,监听topic
    @Bean
    public MessageProducer inbound(MqttPahoClientFactory mqttPahoClientFactory){
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId+"-subscribe",mqttPahoClientFactory,"message/+");
        // 此处设置的是毫秒 10000为十秒
        adapter.setCompletionTimeout(10000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    // 通过通道获取数据
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler receiveHandler(){
        return message -> {
            //MessageHeaders headers = message.getHeaders();
            String m = message.getPayload().toString();
            System.out.println(m);
            mqttService.save(m);  // 处理数据交给具体的业务层
        };
    }

    /************************************************************************************************************/

    // 推送通道
    @Bean
    public MessageChannel mqttOutputChannel(){
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutputChannel")
    public MessageHandler sendHandler(MqttPahoClientFactory mqttPahoClientFactory){
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId+"-publish",mqttPahoClientFactory);
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(1);
        messageHandler.setDefaultTopic(defaultTopic);
        return messageHandler;
    }
}
